/**
 * 
 */
package com.rrd.hive.udaf;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.*;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.io.Text;

import java.util.*;
import java.util.regex.Pattern;


/** 
 * @author baoxin.liu
 * @version 创建时间：2012年04月19日 下午午14:44:43
 * 对神策用户数据进行解析入仓
 * 入参：time,type,distinct_id,properties
 */
public class SensorUserUDAF extends AbstractGenericUDAFResolver {
	public  static  final  String FLAG = "\001\002";
    public static final Pattern CRT001_PATTERN =  Pattern.compile("\001\002");

    @Override
	public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)
			throws SemanticException {
		if (parameters.length == 5) {
			return new AmortizeAmtEvaluator();

		} else {
			throw new UDFArgumentTypeException(parameters.length, "输入参数不正确!");
		}

	}

	public static class AmortizeAmtEvaluator extends GenericUDAFEvaluator {
		private PrimitiveObjectInspector time_inputOI;
		private PrimitiveObjectInspector type_inputOI1;
		private PrimitiveObjectInspector distinct_inputOI2;
		private PrimitiveObjectInspector properties_inputOI3;
        private PrimitiveObjectInspector login_inputOI4;
		private StandardListObjectInspector mapOI;

		private StringBuilder sb = new StringBuilder();
		public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
			super.init(m, parameters);
			if (m == Mode.PARTIAL1) {
                this.time_inputOI = (PrimitiveObjectInspector) parameters[0];
                this.type_inputOI1 = (PrimitiveObjectInspector) parameters[1];
                this.distinct_inputOI2 = (PrimitiveObjectInspector) parameters[2];
                this.properties_inputOI3 = (PrimitiveObjectInspector) parameters[3];
                this.login_inputOI4 = (PrimitiveObjectInspector) parameters[4];
                return ObjectInspectorFactory.getStandardListObjectInspector(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

            } else if (m == Mode.PARTIAL2) {
                this.mapOI = (StandardListObjectInspector) parameters[0];
                return ObjectInspectorFactory.getStandardListObjectInspector(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.FINAL) {
                this.mapOI = (StandardListObjectInspector) parameters[0];
                return ObjectInspectorFactory.getStandardMapObjectInspector(
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.COMPLETE) {
                
                return ObjectInspectorFactory.getStandardMapObjectInspector(
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector,
                        PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else { //
                throw new RuntimeException("no such mode Exception");
            }
		}
		static class UserData extends AbstractAggregationBuffer {
	            List<Text> list = new ArrayList<>();
	     }
		@Override
        public AbstractAggregationBuffer getNewAggregationBuffer()
                throws HiveException {
            // TODO Auto-generated method stub
			AbstractAggregationBuffer user = new UserData();
            this.reset(user);
            return user;
        }
	
        @Override
        public void reset(AggregationBuffer agg) throws HiveException {
            // TODO Auto-generated method stub
            ((UserData) agg).list = new ArrayList<>();

        }
//        public  static final List<String> ONCE_LIST= Arrays.asList(("DownloadChannel,$first_visit_time,$first_referrer,$first_browser_language,$first_browser_charset" +
//                ",$first_traffic_source_type,$first_search_keyword,$first_referrer_host,$utm_source,$utm_medium,$utm_campaign,$utm_content,$utm_term").split(","));
//        public  static final List<String> SET_LIST= Arrays.asList(("UidLast,SourceType,Source").split(","));
        @Override
        public void iterate(AggregationBuffer agg, Object[] parameters)
                throws HiveException {
            if (parameters == null || parameters[0] == null) {
                return;
            }
            UserData userData = (UserData) agg;
            String time = PrimitiveObjectInspectorUtils.getString(parameters[0],this.time_inputOI);
            String type = PrimitiveObjectInspectorUtils.getString(parameters[1],this.type_inputOI1);
            String distinctId = PrimitiveObjectInspectorUtils.getString(parameters[2],this.distinct_inputOI2);
            String propertie = PrimitiveObjectInspectorUtils.getString(parameters[3],this.properties_inputOI3);
            String loginId = PrimitiveObjectInspectorUtils.getString(parameters[4],this.login_inputOI4);
            if(time==null || time.equals("") || type==null || type.equals("")){
                System.out.println("遇到类型或时间为Null的情况，跳过.........");
                return;
            }
            sb.delete(0,sb.length());
            sb.append(time).append(FLAG).append(loginId).append(FLAG).append(distinctId).append(FLAG).append(propertie).append(FLAG).append(type);
            userData.list.add(new Text(sb.toString()));

        }
        @Override
        public void merge(AggregationBuffer agg, Object partial)
                throws HiveException {
            // TODO Auto-generated method stub
            UserData userData = (UserData) agg;
            if (partial != null) { // 在没有输入数据情况下，null会传进来
                List list = this.mapOI.getList(partial);
                userData.list.addAll(list);
            }
        }
        @Override
        public Object terminatePartial(AggregationBuffer agg)
                throws HiveException {
            return ((UserData) agg).list;
        }
        @Override
        public Object terminate(AggregationBuffer agg) throws HiveException {
            UserData user = (UserData) agg;
            if (user.list == null || user.list.size() == 0) {
                // System.out.println("result is null");
                return null;
            }
            Map<String,String> map = new HashMap<>();
            Collections.sort(user.list, new Comparator<Text>() {
                @Override
                public int compare(Text o1, Text o2) {
                    return o1.toString().substring(0,22).compareTo(o2.toString().substring(0,22));
                }
            });
            //time0,lginid1,distinctid2,properties3,type4
            System.out.println("----------");
            for(int i = 0;i<user.list.size();i++){
                String [] arr =CRT001_PATTERN.split(user.list.get(i).toString());
                System.out.println(arr[0]);
                if(arr[4].equals("profile_set_once")){
                    parseJson_once(map,arr[3]);
                }else{
                    parseJson_set(map,arr[3]);
                }
                if(i==0){
                    map.put("firstId",arr[2]);
                }
                if(arr[1]!=null && !arr[1].equals("")){
                    map.put("secondId",arr[1]);
                }

            }
            return map;
        }
        private   void parseJson_once(Map<String,String> map,String properties){
            try{
                JSONObject jsonObject = JSON.parseObject(properties);
                if(!map.containsKey("DownloadChannel") && jsonObject.getString("DownloadChannel")!=null  ){
                    map.put("DownloadChannel",jsonObject.getString("DownloadChannel"));
                }
                if(!map.containsKey("$first_visit_time") && jsonObject.getString("$first_visit_time")!=null){
                    map.put("$first_visit_time",jsonObject.getString("$first_visit_time"));
                }
                if(!map.containsKey("$first_referrer") && jsonObject.getString("$first_referrer")!=null){
                    map.put("$first_referrer",jsonObject.getString("$first_referrer"));
                }
                if(!map.containsKey("$first_browser_language") && jsonObject.getString("$first_browser_language")!=null){
                    map.put("$first_browser_language",jsonObject.getString("$first_browser_language"));
                }
                if(!map.containsKey("$first_browser_charset") && jsonObject.getString("$first_browser_charset")!=null){
                    map.put("$first_browser_charset",jsonObject.getString("$first_browser_charset"));
                }
                if(!map.containsKey("$first_traffic_source_type") && jsonObject.getString("$first_traffic_source_type")!=null){
                    map.put("$first_traffic_source_type",jsonObject.getString("$first_traffic_source_type"));
                }
                if(!map.containsKey("$first_search_keyword") && jsonObject.getString("$first_search_keyword")!=null){
                    map.put("$first_search_keyword",jsonObject.getString("$first_search_keyword"));
                }
                if(!map.containsKey("$first_referrer_host") && jsonObject.getString("$first_referrer_host")!=null){
                    map.put("$first_referrer_host",jsonObject.getString("$first_referrer_host"));
                }
                if(!map.containsKey("$utm_source") && jsonObject.getString("$utm_source")!=null){
                    map.put("$utm_source",jsonObject.getString("$utm_source"));
                }
                if(!map.containsKey("$utm_medium") && jsonObject.getString("$utm_medium")!=null){
                    map.put("$utm_medium",jsonObject.getString("$utm_medium"));
                }
                if(!map.containsKey("$utm_campaign") && jsonObject.getString("$utm_campaign")!=null){
                    map.put("$utm_campaign",jsonObject.getString("$utm_campaign"));
                }
                if(!map.containsKey("$utm_content") && jsonObject.getString("$utm_content")!=null){
                    map.put("$utm_content",jsonObject.getString("$utm_content"));
                }
                if(!map.containsKey("$utm_term") && jsonObject.getString("$utm_term")!=null){
                    map.put("$utm_term",jsonObject.getString("$utm_term"));
                }

            }catch(Exception e){
                e.printStackTrace();
            }
        }
        private void parseJson_set(Map<String,String> map,String properties){
            try{
                JSONObject jsonObject = JSON.parseObject(properties);
                if(jsonObject.getString("UidLast")!=null){
                    map.put("UidLast",jsonObject.getString("UidLast"));
                }
                if(jsonObject.getString("SourceType")!=null){
                    map.put("SourceType",jsonObject.getString("SourceType"));
                }
                if(jsonObject.getString("Source")!=null){
                    map.put("Source",jsonObject.getString("Source"));
                }
            }catch(Exception e){
               e.printStackTrace();
            }
        }
	}

    public static void main(String[] args) {
        System.out.println("2022-02-04".compareTo("2022-05-01"));
    }
}
